use tonic::{Request, Response, Status};
use sea_orm::{DatabaseConnection, EntityTrait, ActiveModelTrait, Set, PaginatorTrait, QueryFilter, ColumnTrait, QueryOrder, ConnectionTrait, TryGetable};
use uuid::Uuid;
use chrono::Utc;
use serde_json::json;
use rust_decimal::Decimal;
use rust_decimal::prelude::ToPrimitive;

use crate::proto::transportation_service_server::TransportationService;
use crate::proto::*;
use crate::entities::{transport_plan, vehicle, order, vehicle_location};
use crate::events::{EventPublisher, EventType};
use crate::services::RouteOptimizer;

pub struct TmsService {
    db: DatabaseConnection,
    event_publisher: EventPublisher,
    route_optimizer: RouteOptimizer,
}

impl TmsService {
    pub fn new(db: DatabaseConnection, event_publisher: EventPublisher) -> Self {
        Self {
            db,
            event_publisher,
            route_optimizer: RouteOptimizer::new(),
        }
    }
}

#[tonic::async_trait]
impl TransportationService for TmsService {
    async fn create_transport_plan(
        &self,
        request: Request<CreateTransportPlanRequest>,
    ) -> Result<Response<TransportPlan>, Status> {
        let req = request.into_inner();
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().naive_utc();

        let plan = transport_plan::ActiveModel {
            id: Set(id.clone()),
            name: Set(req.name.clone()),
            description: Set(if req.description.is_empty() { None } else { Some(req.description.clone()) }),
            order_ids: Set(serde_json::to_string(&req.order_ids).unwrap()),
            vehicle_id: Set(if req.vehicle_id.is_empty() { None } else { Some(req.vehicle_id.clone()) }),
            driver_id: Set(if req.driver_id.is_empty() { None } else { Some(req.driver_id.clone()) }),
            route_data: Set(None),
            planned_start_time: Set(if req.planned_start_time == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.planned_start_time, 0).unwrap().naive_utc()) 
            }),
            planned_end_time: Set(if req.planned_end_time == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.planned_end_time, 0).unwrap().naive_utc()) 
            }),
            status: Set(0), // pending
            created_at: Set(now),
            updated_at: Set(now),
        };

        // Use raw SQL insert to avoid SeaORM's save/update logic
        use sea_orm::Statement;
        let insert_sql = Statement::from_string(
            sea_orm::DatabaseBackend::MySql,
            format!(
                "INSERT INTO transport_plans (id, name, description, order_ids, vehicle_id, driver_id, route_data, planned_start_time, planned_end_time, status, created_at, updated_at) VALUES ('{}', '{}', {}, '{}', {}, {}, {}, {}, {}, {}, '{}', '{}')",
                id,
                req.name.replace("'", "''"), // Escape single quotes
                if req.description.is_empty() { "NULL".to_string() } else { format!("'{}'", req.description.replace("'", "''")) },
                serde_json::to_string(&req.order_ids).unwrap(),
                if req.vehicle_id.is_empty() { "NULL".to_string() } else { format!("'{}'", req.vehicle_id) },
                if req.driver_id.is_empty() { "NULL".to_string() } else { format!("'{}'", req.driver_id) },
                "NULL", // route_data
                if req.planned_start_time == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.planned_start_time, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                if req.planned_end_time == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.planned_end_time, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                0, // pending
                now.format("%Y-%m-%d %H:%M:%S"),
                now.format("%Y-%m-%d %H:%M:%S")
            )
        );
        
        let insert_result = self.db.execute(insert_sql).await
            .map_err(|e| Status::internal(format!("Failed to create transport plan: {}", e)))?;

        if insert_result.rows_affected() == 0 {
            return Err(Status::internal("Failed to insert transport plan"));
        }

        // Get the inserted record using raw SQL to avoid JSON type issues
        let query_sql = Statement::from_string(
            sea_orm::DatabaseBackend::MySql,
            format!("SELECT id, name, description, order_ids, vehicle_id, driver_id, route_data, planned_start_time, planned_end_time, status, created_at, updated_at FROM transport_plans WHERE id = '{}'", id)
        );
        
        let query_result = self.db.query_one(query_sql).await
            .map_err(|e| Status::internal(format!("Failed to fetch created transport plan: {}", e)))?
            .ok_or_else(|| Status::internal("Created transport plan not found"))?;

        // Manually construct the result
        let result = transport_plan::Model {
            id: query_result.try_get("", "id").unwrap_or_default(),
            name: query_result.try_get("", "name").unwrap_or_default(),
            description: query_result.try_get("", "description").ok(),
            order_ids: query_result.try_get("", "order_ids").unwrap_or_default(),
            vehicle_id: query_result.try_get("", "vehicle_id").ok(),
            driver_id: query_result.try_get("", "driver_id").ok(),
            route_data: query_result.try_get("", "route_data").ok(),
            planned_start_time: query_result.try_get("", "planned_start_time").ok(),
            planned_end_time: query_result.try_get("", "planned_end_time").ok(),
            status: query_result.try_get("", "status").unwrap_or_default(),
            created_at: query_result.try_get("", "created_at").unwrap_or_default(),
            updated_at: query_result.try_get("", "updated_at").unwrap_or_default(),
        };

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "name": result.name,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_transport_plan_event(
            &result.id,
            EventType::Created,
            event_data,
        ).await {
            tracing::warn!("Failed to publish transport plan created event: {}", e);
        }

        let response = TransportPlan {
            id: result.id,
            name: result.name,
            description: result.description.unwrap_or_default(),
            order_ids: serde_json::from_str(&result.order_ids).unwrap_or_default(),
            vehicle_id: result.vehicle_id.unwrap_or_default(),
            driver_id: result.driver_id.unwrap_or_default(),
            route_data: result.route_data.unwrap_or_default(),
            planned_start_time: result.planned_start_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            planned_end_time: result.planned_end_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            status: result.status,
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn get_transport_plan(
        &self,
        request: Request<GetTransportPlanRequest>,
    ) -> Result<Response<TransportPlan>, Status> {
        let req = request.into_inner();
        
        let plan = transport_plan::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Transport plan not found"))?;

        let response = TransportPlan {
            id: plan.id,
            name: plan.name,
            description: plan.description.unwrap_or_default(),
            order_ids: serde_json::from_str(&plan.order_ids).unwrap_or_default(),
            vehicle_id: plan.vehicle_id.unwrap_or_default(),
            driver_id: plan.driver_id.unwrap_or_default(),
            route_data: plan.route_data.unwrap_or_default(),
            planned_start_time: plan.planned_start_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            planned_end_time: plan.planned_end_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            status: plan.status,
            created_at: plan.created_at.and_utc().timestamp(),
            updated_at: plan.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn list_transport_plans(
        &self,
        request: Request<ListTransportPlansRequest>,
    ) -> Result<Response<ListTransportPlansResponse>, Status> {
        let req = request.into_inner();
        let page = req.page.max(1) as u64;
        let page_size = req.page_size.min(100).max(1) as u64;

        let mut query = transport_plan::Entity::find();
        
        if req.status != 0 {
            query = query.filter(transport_plan::Column::Status.eq(req.status));
        }

        let paginator = query.paginate(&self.db, page_size);
        let total = paginator.num_items().await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let plans = paginator.fetch_page(page - 1).await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let response_plans: Vec<TransportPlan> = plans.into_iter().map(|plan| {
            TransportPlan {
                id: plan.id,
                name: plan.name,
                description: plan.description.unwrap_or_default(),
                order_ids: serde_json::from_str(&plan.order_ids).unwrap_or_default(),
                vehicle_id: plan.vehicle_id.unwrap_or_default(),
                driver_id: plan.driver_id.unwrap_or_default(),
                route_data: plan.route_data.unwrap_or_default(),
                planned_start_time: plan.planned_start_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                planned_end_time: plan.planned_end_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                status: plan.status,
                created_at: plan.created_at.and_utc().timestamp(),
                updated_at: plan.updated_at.and_utc().timestamp(),
            }
        }).collect();

        let response = ListTransportPlansResponse {
            plans: response_plans,
            total: total as i32,
        };

        Ok(Response::new(response))
    }

    async fn update_transport_plan(
        &self,
        request: Request<UpdateTransportPlanRequest>,
    ) -> Result<Response<TransportPlan>, Status> {
        let req = request.into_inner();
        
        let plan = transport_plan::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Transport plan not found"))?;

        let mut active_plan: transport_plan::ActiveModel = plan.into();
        
        if let Some(name) = req.name {
            active_plan.name = Set(name);
        }
        if let Some(description) = req.description {
            active_plan.description = Set(Some(description));
        }
        if !req.order_ids.is_empty() {
            active_plan.order_ids = Set(serde_json::to_string(&req.order_ids).unwrap());
        }
        if let Some(vehicle_id) = req.vehicle_id {
            active_plan.vehicle_id = Set(Some(vehicle_id));
        }
        if let Some(driver_id) = req.driver_id {
            active_plan.driver_id = Set(Some(driver_id));
        }
        if let Some(start_time) = req.planned_start_time {
            active_plan.planned_start_time = Set(Some(chrono::DateTime::from_timestamp(start_time, 0).unwrap().naive_utc()));
        }
        if let Some(end_time) = req.planned_end_time {
            active_plan.planned_end_time = Set(Some(chrono::DateTime::from_timestamp(end_time, 0).unwrap().naive_utc()));
        }
        if let Some(status) = req.status {
            active_plan.status = Set(status);
        }
        
        active_plan.updated_at = Set(Utc::now().naive_utc());

        let result = active_plan.update(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to update transport plan: {}", e)))?;

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "name": result.name,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_transport_plan_event(
            &result.id,
            EventType::Updated,
            event_data,
        ).await {
            tracing::warn!("Failed to publish transport plan updated event: {}", e);
        }

        let response = TransportPlan {
            id: result.id,
            name: result.name,
            description: result.description.unwrap_or_default(),
            order_ids: serde_json::from_str(&result.order_ids).unwrap_or_default(),
            vehicle_id: result.vehicle_id.unwrap_or_default(),
            driver_id: result.driver_id.unwrap_or_default(),
            route_data: result.route_data.unwrap_or_default(),
            planned_start_time: result.planned_start_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            planned_end_time: result.planned_end_time.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            status: result.status,
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn delete_transport_plan(
        &self,
        request: Request<DeleteTransportPlanRequest>,
    ) -> Result<Response<DeleteTransportPlanResponse>, Status> {
        let req = request.into_inner();
        
        let result = transport_plan::Entity::delete_by_id(&req.id)
            .exec(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Failed to delete transport plan: {}", e)))?;

        if result.rows_affected == 0 {
            return Err(Status::not_found("Transport plan not found"));
        }

        // 发布事件
        let event_data = json!({
            "id": req.id
        });
        
        if let Err(e) = self.event_publisher.publish_transport_plan_event(
            &req.id,
            EventType::Deleted,
            event_data,
        ).await {
            tracing::warn!("Failed to publish transport plan deleted event: {}", e);
        }

        let response = DeleteTransportPlanResponse { success: true };
        Ok(Response::new(response))
    }

    // Vehicle methods
    async fn create_vehicle(
        &self,
        request: Request<CreateVehicleRequest>,
    ) -> Result<Response<Vehicle>, Status> {
        let req = request.into_inner();
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().naive_utc();

        // 先检查是否已存在相同的车牌号
        let existing = vehicle::Entity::find()
            .filter(vehicle::Column::LicensePlate.eq(&req.license_plate))
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;
        
        if existing.is_some() {
            return Err(Status::already_exists("Vehicle with this license plate already exists"));
        }

        // Use raw SQL insert to avoid SeaORM's save/update logic
        use sea_orm::Statement;
        let insert_sql = Statement::from_string(
            sea_orm::DatabaseBackend::MySql,
            format!(
                "INSERT INTO vehicles (id, license_plate, model, max_weight, max_volume, vehicle_type, status, driver_id, created_at, updated_at) VALUES ('{}', '{}', '{}', {}, {}, {}, {}, {}, '{}', '{}')",
                id,
                req.license_plate,
                req.model,
                Decimal::from_f64_retain(req.max_weight).unwrap_or_default(),
                Decimal::from_f64_retain(req.max_volume).unwrap_or_default(),
                req.r#type,
                0,
                if req.driver_id.is_empty() { "NULL".to_string() } else { format!("'{}'", req.driver_id) },
                now.format("%Y-%m-%d %H:%M:%S"),
                now.format("%Y-%m-%d %H:%M:%S")
            )
        );
        
        let insert_result = self.db.execute(insert_sql).await
            .map_err(|e| Status::internal(format!("Failed to create vehicle: {}", e)))?;

        if insert_result.rows_affected() == 0 {
            return Err(Status::internal("Failed to insert vehicle"));
        }
        
        // Get the inserted record
        let result = vehicle::Entity::find_by_id(&id).one(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to fetch created vehicle: {}", e)))?
            .ok_or_else(|| Status::internal("Created vehicle not found"))?;

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "license_plate": result.license_plate,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_vehicle_event(
            &result.id,
            EventType::Created,
            event_data,
        ).await {
            tracing::warn!("Failed to publish vehicle created event: {}", e);
        }

        let response = Vehicle {
            id: result.id,
            license_plate: result.license_plate,
            model: result.model,
            max_weight: result.max_weight.to_f64().unwrap_or_default(),
            max_volume: result.max_volume.to_f64().unwrap_or_default(),
            r#type: result.vehicle_type,
            status: result.status,
            driver_id: result.driver_id.unwrap_or_default(),
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn get_vehicle(
        &self,
        request: Request<GetVehicleRequest>,
    ) -> Result<Response<Vehicle>, Status> {
        let req = request.into_inner();
        
        let vehicle = vehicle::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Vehicle not found"))?;

        let response = Vehicle {
            id: vehicle.id,
            license_plate: vehicle.license_plate,
            model: vehicle.model,
            max_weight: vehicle.max_weight.to_f64().unwrap_or_default(),
            max_volume: vehicle.max_volume.to_f64().unwrap_or_default(),
            r#type: vehicle.vehicle_type,
            status: vehicle.status,
            driver_id: vehicle.driver_id.unwrap_or_default(),
            created_at: vehicle.created_at.and_utc().timestamp(),
            updated_at: vehicle.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn list_vehicles(
        &self,
        request: Request<ListVehiclesRequest>,
    ) -> Result<Response<ListVehiclesResponse>, Status> {
        let req = request.into_inner();
        let page = req.page.max(1) as u64;
        let page_size = req.page_size.min(100).max(1) as u64;

        let mut query = vehicle::Entity::find();
        
        if req.status != 0 {
            query = query.filter(vehicle::Column::Status.eq(req.status));
        }

        let paginator = query.paginate(&self.db, page_size);
        let total = paginator.num_items().await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let vehicles = paginator.fetch_page(page - 1).await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let response_vehicles: Vec<Vehicle> = vehicles.into_iter().map(|vehicle| {
            Vehicle {
                id: vehicle.id,
                license_plate: vehicle.license_plate,
                model: vehicle.model,
                max_weight: vehicle.max_weight.to_f64().unwrap_or_default(),
                max_volume: vehicle.max_volume.to_f64().unwrap_or_default(),
                r#type: vehicle.vehicle_type,
                status: vehicle.status,
                driver_id: vehicle.driver_id.unwrap_or_default(),
                created_at: vehicle.created_at.and_utc().timestamp(),
                updated_at: vehicle.updated_at.and_utc().timestamp(),
            }
        }).collect();

        let response = ListVehiclesResponse {
            vehicles: response_vehicles,
            total: total as i32,
        };

        Ok(Response::new(response))
    }

    async fn update_vehicle(
        &self,
        request: Request<UpdateVehicleRequest>,
    ) -> Result<Response<Vehicle>, Status> {
        let req = request.into_inner();
        
        let vehicle = vehicle::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Vehicle not found"))?;

        let mut active_vehicle: vehicle::ActiveModel = vehicle.into();
        
        if let Some(license_plate) = req.license_plate {
            active_vehicle.license_plate = Set(license_plate);
        }
        if let Some(model) = req.model {
            active_vehicle.model = Set(model);
        }
        if let Some(max_weight) = req.max_weight {
            active_vehicle.max_weight = Set(Decimal::from_f64_retain(max_weight).unwrap_or_default());
        }
        if let Some(max_volume) = req.max_volume {
            active_vehicle.max_volume = Set(Decimal::from_f64_retain(max_volume).unwrap_or_default());
        }
        if let Some(vehicle_type) = req.r#type {
            active_vehicle.vehicle_type = Set(vehicle_type);
        }
        if let Some(status) = req.status {
            active_vehicle.status = Set(status);
        }
        if let Some(driver_id) = req.driver_id {
            active_vehicle.driver_id = Set(Some(driver_id));
        }
        
        active_vehicle.updated_at = Set(Utc::now().naive_utc());

        let result = active_vehicle.update(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to update vehicle: {}", e)))?;

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "license_plate": result.license_plate,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_vehicle_event(
            &result.id,
            EventType::Updated,
            event_data,
        ).await {
            tracing::warn!("Failed to publish vehicle updated event: {}", e);
        }

        let response = Vehicle {
            id: result.id,
            license_plate: result.license_plate,
            model: result.model,
            max_weight: result.max_weight.to_f64().unwrap_or_default(),
            max_volume: result.max_volume.to_f64().unwrap_or_default(),
            r#type: result.vehicle_type,
            status: result.status,
            driver_id: result.driver_id.unwrap_or_default(),
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    // Order methods
    async fn create_order(
        &self,
        request: Request<CreateOrderRequest>,
    ) -> Result<Response<Order>, Status> {
        let req = request.into_inner();
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().naive_utc();

        let order = order::ActiveModel {
            id: Set(id.clone()),
            customer_id: Set(req.customer_id.clone()),
            pickup_address: Set(req.pickup_address.clone()),
            delivery_address: Set(req.delivery_address.clone()),
            weight: Set(Decimal::from_f64_retain(req.weight).unwrap_or_default()),
            volume: Set(Decimal::from_f64_retain(req.volume).unwrap_or_default()),
            status: Set(0), // pending
            pickup_time_window_start: Set(if req.pickup_time_window_start == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.pickup_time_window_start, 0).unwrap().naive_utc()) 
            }),
            pickup_time_window_end: Set(if req.pickup_time_window_end == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.pickup_time_window_end, 0).unwrap().naive_utc()) 
            }),
            delivery_time_window_start: Set(if req.delivery_time_window_start == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.delivery_time_window_start, 0).unwrap().naive_utc()) 
            }),
            delivery_time_window_end: Set(if req.delivery_time_window_end == 0 { None } else { 
                Some(chrono::DateTime::from_timestamp(req.delivery_time_window_end, 0).unwrap().naive_utc()) 
            }),
            special_requirements: Set(if req.special_requirements.is_empty() { None } else { Some(req.special_requirements.clone()) }),
            created_at: Set(now),
            updated_at: Set(now),
        };

        // Use raw SQL insert to avoid SeaORM's save/update logic
        use sea_orm::Statement;
        let insert_sql = Statement::from_string(
            sea_orm::DatabaseBackend::MySql,
            format!(
                "INSERT INTO orders (id, customer_id, pickup_address, delivery_address, weight, volume, status, pickup_time_window_start, pickup_time_window_end, delivery_time_window_start, delivery_time_window_end, special_requirements, created_at, updated_at) VALUES ('{}', '{}', '{}', '{}', {}, {}, {}, {}, {}, {}, {}, {}, '{}', '{}')",
                id,
                req.customer_id.replace("'", "''"),
                req.pickup_address.replace("'", "''"),
                req.delivery_address.replace("'", "''"),
                Decimal::from_f64_retain(req.weight).unwrap_or_default(),
                Decimal::from_f64_retain(req.volume).unwrap_or_default(),
                0, // pending
                if req.pickup_time_window_start == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.pickup_time_window_start, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                if req.pickup_time_window_end == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.pickup_time_window_end, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                if req.delivery_time_window_start == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.delivery_time_window_start, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                if req.delivery_time_window_end == 0 { "NULL".to_string() } else { 
                    format!("'{}'", chrono::DateTime::from_timestamp(req.delivery_time_window_end, 0).unwrap().naive_utc().format("%Y-%m-%d %H:%M:%S"))
                },
                if req.special_requirements.is_empty() { "NULL".to_string() } else { format!("'{}'", req.special_requirements.replace("'", "''")) },
                now.format("%Y-%m-%d %H:%M:%S"),
                now.format("%Y-%m-%d %H:%M:%S")
            )
        );
        
        let insert_result = self.db.execute(insert_sql).await
            .map_err(|e| Status::internal(format!("Failed to create order: {}", e)))?;

        if insert_result.rows_affected() == 0 {
            return Err(Status::internal("Failed to insert order"));
        }

        // Get the inserted record
        let result = order::Entity::find_by_id(&id).one(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to fetch created order: {}", e)))?
            .ok_or_else(|| Status::internal("Created order not found"))?;

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "customer_id": result.customer_id,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_order_event(
            &result.id,
            EventType::Created,
            event_data,
        ).await {
            tracing::warn!("Failed to publish order created event: {}", e);
        }

        let response = Order {
            id: result.id,
            customer_id: result.customer_id,
            pickup_address: result.pickup_address,
            delivery_address: result.delivery_address,
            weight: result.weight.to_f64().unwrap_or_default(),
            volume: result.volume.to_f64().unwrap_or_default(),
            status: result.status,
            pickup_time_window_start: result.pickup_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            pickup_time_window_end: result.pickup_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_start: result.delivery_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_end: result.delivery_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            special_requirements: result.special_requirements.unwrap_or_default(),
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn get_order(
        &self,
        request: Request<GetOrderRequest>,
    ) -> Result<Response<Order>, Status> {
        let req = request.into_inner();
        
        let order = order::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Order not found"))?;

        let response = Order {
            id: order.id,
            customer_id: order.customer_id,
            pickup_address: order.pickup_address,
            delivery_address: order.delivery_address,
            weight: order.weight.to_f64().unwrap_or_default(),
            volume: order.volume.to_f64().unwrap_or_default(),
            status: order.status,
            pickup_time_window_start: order.pickup_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            pickup_time_window_end: order.pickup_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_start: order.delivery_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_end: order.delivery_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            special_requirements: order.special_requirements.unwrap_or_default(),
            created_at: order.created_at.and_utc().timestamp(),
            updated_at: order.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn list_orders(
        &self,
        request: Request<ListOrdersRequest>,
    ) -> Result<Response<ListOrdersResponse>, Status> {
        let req = request.into_inner();
        let page = req.page.max(1) as u64;
        let page_size = req.page_size.min(100).max(1) as u64;

        let mut query = order::Entity::find();
        
        if req.status != 0 {
            query = query.filter(order::Column::Status.eq(req.status));
        }

        let paginator = query.paginate(&self.db, page_size);
        let total = paginator.num_items().await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let orders = paginator.fetch_page(page - 1).await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        let response_orders: Vec<Order> = orders.into_iter().map(|order| {
            Order {
                id: order.id,
                customer_id: order.customer_id,
                pickup_address: order.pickup_address,
                delivery_address: order.delivery_address,
                weight: order.weight.to_f64().unwrap_or_default(),
                volume: order.volume.to_f64().unwrap_or_default(),
                status: order.status,
                pickup_time_window_start: order.pickup_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                pickup_time_window_end: order.pickup_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                delivery_time_window_start: order.delivery_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                delivery_time_window_end: order.delivery_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
                special_requirements: order.special_requirements.unwrap_or_default(),
                created_at: order.created_at.and_utc().timestamp(),
                updated_at: order.updated_at.and_utc().timestamp(),
            }
        }).collect();

        let response = ListOrdersResponse {
            orders: response_orders,
            total: total as i32,
        };

        Ok(Response::new(response))
    }

    async fn update_order_status(
        &self,
        request: Request<UpdateOrderStatusRequest>,
    ) -> Result<Response<Order>, Status> {
        let req = request.into_inner();
        
        let order = order::Entity::find_by_id(&req.id)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Order not found"))?;

        let mut active_order: order::ActiveModel = order.into();
        active_order.status = Set(req.status);
        active_order.updated_at = Set(Utc::now().naive_utc());

        let result = active_order.update(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to update order: {}", e)))?;

        // 发布事件
        let event_data = json!({
            "id": result.id,
            "customer_id": result.customer_id,
            "status": result.status
        });
        
        if let Err(e) = self.event_publisher.publish_order_event(
            &result.id,
            EventType::StatusChanged,
            event_data,
        ).await {
            tracing::warn!("Failed to publish order status changed event: {}", e);
        }

        let response = Order {
            id: result.id,
            customer_id: result.customer_id,
            pickup_address: result.pickup_address,
            delivery_address: result.delivery_address,
            weight: result.weight.to_f64().unwrap_or_default(),
            volume: result.volume.to_f64().unwrap_or_default(),
            status: result.status,
            pickup_time_window_start: result.pickup_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            pickup_time_window_end: result.pickup_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_start: result.delivery_time_window_start.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            delivery_time_window_end: result.delivery_time_window_end.map(|t| t.and_utc().timestamp()).unwrap_or(0),
            special_requirements: result.special_requirements.unwrap_or_default(),
            created_at: result.created_at.and_utc().timestamp(),
            updated_at: result.updated_at.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }

    async fn optimize_route(
        &self,
        request: Request<OptimizeRouteRequest>,
    ) -> Result<Response<OptimizeRouteResponse>, Status> {
        let req = request.into_inner();
        
        // 获取订单信息
        let orders = order::Entity::find()
            .filter(order::Column::Id.is_in(req.order_ids.clone()))
            .all(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?;

        if orders.is_empty() {
            return Err(Status::invalid_argument("No valid orders found"));
        }

        // 使用路线优化器
        let optimization_result = self.route_optimizer.optimize_route(
            &orders,
            req.objective(),
        ).await
            .map_err(|e| Status::internal(format!("Route optimization failed: {}", e)))?;

        let response = OptimizeRouteResponse {
            optimized_order_sequence: optimization_result.order_sequence,
            total_distance: optimization_result.total_distance,
            estimated_time: optimization_result.estimated_time,
            estimated_cost: optimization_result.estimated_cost,
        };

        Ok(Response::new(response))
    }

    async fn update_location(
        &self,
        request: Request<UpdateLocationRequest>,
    ) -> Result<Response<UpdateLocationResponse>, Status> {
        let req = request.into_inner();
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().naive_utc();

        let location = vehicle_location::ActiveModel {
            id: Set(id),
            vehicle_id: Set(req.vehicle_id.clone()),
            latitude: Set(Decimal::from_f64_retain(req.latitude).unwrap_or_default()),
            longitude: Set(Decimal::from_f64_retain(req.longitude).unwrap_or_default()),
            speed: Set(Decimal::from_f64_retain(req.speed).unwrap_or_default()),
            heading: Set(Decimal::from_f64_retain(req.heading).unwrap_or_default()),
            timestamp: Set(now),
            created_at: Set(now),
        };

        vehicle_location::Entity::insert(location).exec(&self.db).await
            .map_err(|e| Status::internal(format!("Failed to update location: {}", e)))?;

        // 发布位置事件
        let event_data = json!({
            "vehicle_id": req.vehicle_id,
            "latitude": req.latitude,
            "longitude": req.longitude,
            "speed": req.speed,
            "heading": req.heading,
            "timestamp": now.and_utc().timestamp()
        });
        
        if let Err(e) = self.event_publisher.publish_location_event(
            &req.vehicle_id,
            event_data,
        ).await {
            tracing::warn!("Failed to publish location event: {}", e);
        }

        let response = UpdateLocationResponse { success: true };
        Ok(Response::new(response))
    }

    async fn get_vehicle_location(
        &self,
        request: Request<GetVehicleLocationRequest>,
    ) -> Result<Response<VehicleLocation>, Status> {
        let req = request.into_inner();
        
        let location = vehicle_location::Entity::find()
            .filter(vehicle_location::Column::VehicleId.eq(&req.vehicle_id))
            .order_by_desc(vehicle_location::Column::Timestamp)
            .one(&self.db)
            .await
            .map_err(|e| Status::internal(format!("Database error: {}", e)))?
            .ok_or_else(|| Status::not_found("Vehicle location not found"))?;

        let response = VehicleLocation {
            vehicle_id: location.vehicle_id,
            latitude: location.latitude.to_f64().unwrap_or_default(),
            longitude: location.longitude.to_f64().unwrap_or_default(),
            speed: location.speed.to_f64().unwrap_or_default(),
            heading: location.heading.to_f64().unwrap_or_default(),
            timestamp: location.timestamp.and_utc().timestamp(),
        };

        Ok(Response::new(response))
    }
}